package com.ruoyi.job.kettle;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KettleQuartz {
	protected static final Logger logger_info = LoggerFactory.getLogger(KettleQuartz.class);

	public static void main(String[] args) {
		Map<String, String> map = new HashMap<String, String>();
//		map.put("qmcy_path","D:/涟水/涟水精准营销/");
		runKettleTransfer(null, "D:\\涟水\\涟水精准营销\\XXX\\hive\\QMCY_CUST_SJYH.ktr");
//		runKettleJob(map, "D:\\涟水\\涟水精准营销\\XXX\\JXXT.kjb");
	}

	/**
	 * 执行作业
	 * 
	 * @param initKettleParam
	 * @param kjbFilePath
	 * @return
	 */
	public static boolean runKettleJob(Map<String, String> initKettleParam, String kjbFilePath) {
		String uuid = UUID.randomUUID().toString();
		logger_info.info("ExecKettleUtil@runKettleJob:" + uuid + " {kjbFilePath:" + kjbFilePath + "}");
		try {
			KettleEnvironment.init();
			// 初始化job路径
			JobMeta jobMeta = new JobMeta(kjbFilePath, null);
			Job job = new Job(null, jobMeta);
			// 初始化job参数，脚本中获取参数值：${variableName}
			if (initKettleParam != null) {
				for (String variableName : initKettleParam.keySet()) {
					job.setVariable(variableName, initKettleParam.get(variableName));
				}
			}

			job.start();
			job.waitUntilFinished();
			if (job.getErrors() > 0) {
				logger_info.info("ExecKettleUtil@runKettleJob:" + uuid + " 执行失败");
			} else {
				logger_info.info("ExecKettleUtil@runKettleJob:" + uuid + " 执行成功");
			}
			return true;
		} catch (Exception e) {
			logger_info.error("ExecKettleUtil@runKettleJob:" + uuid, e);
			return false;
		}
	}

	/**
	 * 执行转换 执行转换的ktr中，输入表的字段不能as，需要经过 “字段选择” 组件进行转换，否则报错 参考
	 * D:\\涟水\\涟水精准营销\\XXX\\JXXT.ktr
	 * 
	 * @param initKettleParam
	 * @param ktrFilePath
	 * @return
	 */
	public static boolean runKettleTransfer(Map<String, String> initKettleParam, String ktrFilePath) {
		Trans trans = null;
		String uuid = UUID.randomUUID().toString();
		logger_info.info("ExecKettleUtil@runKettleTransfer:" + uuid + " {ktrFilePath:" + ktrFilePath + "}");
		try {
			// 初始化
			KettleEnvironment.init();
			EnvUtil.environmentInit();
			TransMeta transMeta = new TransMeta(ktrFilePath);
			// 转换
			trans = new Trans(transMeta);
			// 初始化trans参数，脚本中获取参数值：${variableName}
			if (initKettleParam != null) {
				for (String variableName : initKettleParam.keySet()) {
					trans.setVariable(variableName, initKettleParam.get(variableName));
				}
			}

			// 执行转换
			trans.execute(null);
			// 等待转换执行结束
			trans.waitUntilFinished();
			
			if (trans.getErrors() > 0) {
				logger_info.info("ExecKettleUtil@runKettleTransfer:" + uuid + " 执行失败");
			} else {
				logger_info.info("ExecKettleUtil@runKettleTransfer:" + uuid + " 执行成功");
			}
			return true;
		} catch (Exception e) {
			logger_info.error("ExecKettleUtil@runKettleTransfer:" + uuid, e);
			return false;
		}
	}

}
